package main

import (
	"context"
	"fmt"
	"github.com/google/uuid"
	amqp "github.com/rabbitmq/amqp091-go"
	"my_go_open/utils"
	"time"
)

var mqClient *utils.AmqpClient

func init() {
	var err error
	mqClient, err = utils.NewAmqpClient("amqp://root:abc123@localhost:5672/")
	if err != nil {
		utils.VarDump("amqp dial err", err.Error())
		return
	}
	ch, _ := mqClient.GetChannel()
	defer mqClient.PutChannel(ch)
	_, _ = ch.QueueDeclare("request", false, false, false, false, nil)
	_, _ = ch.QueueDeclare("response", false, false, false, false, nil)
}

func sendMsg() {
	ch, _ := mqClient.GetChannel()
	defer mqClient.PutChannel(ch)
	cid := uuid.New().String()
	if err := ch.PublishWithContext(context.Background(), "", "request", false, false, amqp.Publishing{
		ContentType:   "text/plain",
		CorrelationId: cid,
		ReplyTo:       "response",
		Timestamp:     time.Now(),
		Body:          []byte(utils.TimeNowNice()),
	}); err != nil {
		utils.VarDump("send err", err.Error())
	}
	ms, err := ch.Consume("response", "", false, false, false, false, nil)
	if err != nil {
		utils.VarDump("client consume err", err.Error())
		return
	}
	stop := false
	for !stop {
		select {
		case m := <-ms:
			if m.CorrelationId == cid {
				utils.VarDump("rpc response:", string(m.Body))
				stop = true
				if err = m.Ack(false); err != nil {
					utils.VarDump("client ack err", err.Error())
				}
				break
			}
			if m.Timestamp.Add(10 * time.Second).Before(time.Now()) {
				if err = m.Ack(false); err != nil {
					utils.VarDump("client drop msg, client ack err", err.Error())
				}
			}
		}
	}
}

func replyMsg() {
	ch, _ := mqClient.GetChannel()
	defer mqClient.PutChannel(ch)
	ms, _ := ch.Consume("request", "", false, false, false, false, nil)
	for {
		select {
		case m := <-ms:
			if err := m.Ack(false); err != nil {
				utils.VarDump("server ack err", err.Error())
			}
			if m.Timestamp.Add(10 * time.Second).Before(time.Now()) {
				continue
			}
			if err := ch.PublishWithContext(context.Background(), "", m.ReplyTo, false, false, amqp.Publishing{
				ContentType:   "text/plain",
				CorrelationId: m.CorrelationId,
				Body:          []byte(fmt.Sprintf("%s, got", string(m.Body))),
				Timestamp:     time.Now(),
			}); err != nil {
				utils.VarDump("server publish err", err.Error())
			}
		}
	}
}

func main() {
	go sendMsg()
	go replyMsg()
	time.Sleep(3 * time.Second)
}
